/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.jr.swb.rest.mq.handlers;

import cn.ac.iie.di.ban.jr.swb.rest.mq.global.GlobalParas;
import cn.ac.iie.di.ban.jr.swb.rest.mq.utils.DateUtils;
import cn.ac.iie.di.ban.jr.swb.rest.mq.utils.FileAppender;
import cn.ac.iie.di.ban.jr.swb.rest.mq.utils.SendData;
import cn.ac.iie.di.ban.jr.swb.rest.mq.utils.SenderPool;
import cn.ac.iie.di.ban.jr.swb.rest.mq.utils.utils;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import com.alibaba.fastjson.JSONObject;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class Hotwb_Handler {

    private String dataType;

    public static Logger logger = Logger.getLogger(Hotwb_Handler.class);
    private SenderPool pool;
    private FileAppender appender;
    private List<String> column_name_list;
    private List<REFieldType> column_type_list;

    public Hotwb_Handler(String dataType, SenderPool pool, FileAppender appender, List<String> column_name_list, List<REFieldType> column_type_list) {
        this.dataType = dataType;
        this.pool = pool;
        this.appender = appender;
        this.column_name_list = column_name_list;
        this.column_type_list = column_type_list;
    }

    public void execute() {
        //根据dataType获取目录
        try {

            String file_dir = GlobalParas.FILE_DIR + dataType + "/";
            //轮询昨日和今日文件夹
            ArrayList<File> shoulddoList = new ArrayList<>();

            //轮询分类底下所有日期文件夹
            File filedir = new File(file_dir);
            File[] files = filedir.listFiles();
            for (File file : files) {
                if (file.isDirectory()) {
                    logger.info("into directory " + file.getAbsolutePath());
                    ArrayList<File> tempList = getFile(file);
                    for (File tempList1 : tempList) {
                        shoulddoList.add(tempList1);
                    }
                } else {
                    logger.warn("this is a file" + file.getAbsolutePath());
                }
            }
            if (shoulddoList.size() > 0) {
                for (File file : shoulddoList) {
                    logger.info("start do fileName = " + file.getAbsolutePath());
                    FileReader reader = null;
                    try {
                        //已经完成的文件
                        String obj_gaterTime = "";
                        String fileName = file.getAbsolutePath();
                        reader = new FileReader(file);
                        logger.debug(fileName);
                        BufferedReader br = new BufferedReader(reader);
                        while ((obj_gaterTime = br.readLine()) != null) {
                            if (obj_gaterTime.contains("}|")) {
                                String[] objgaterTime = obj_gaterTime.split("}\\|");
                                String obj = objgaterTime[0] + "}";
                                Long gaterTime = Long.parseLong(objgaterTime[1]);
                                logger.debug(dataType + " json = " + obj_gaterTime);
                                //判断是否是正常json
                                try {
                                    JSONObject objJson = JSONObject.parseObject(obj);
                                } catch (Exception e) {
                                    logger.error("objJson is not a json string " + e.getMessage(), e);
                                    continue;
                                }

                                //发送消息队列
                                logger.info("start to send queue!!!===============================================");
                                RESendSession sender = null;
                                try {
                                    try {
                                        logger.info("send dataType ：" + dataType + ",topic:" + GlobalParas.CONSUMER_DATATYPE_TOPIC_MAP.get(dataType) + ",column name lists are :" + column_name_list);
                                        List<Object[]> datas = utils.getTHotData(dataType, JSONObject.parseObject(obj), column_name_list, gaterTime, -1);
                                        if (true) {
                                            SendData sendData = new SendData(GlobalParas.CONSUMER_DATATYPE_TOPIC_MAP.get(dataType));
                                            sendData.sendMessage(dataType, column_name_list, column_type_list.toArray(new REFieldType[column_type_list.size()]), datas);
                                        }
                                    } catch (ConfigurationException e) {
                                        logger.error(ExceptionUtils.getFullStackTrace(e));
                                    }

                                } catch (RESessionException ex) {
                                    logger.error(ExceptionUtils.getFullStackTrace(ex));
                                } catch (REConnectionException ex) {
                                    logger.error(ExceptionUtils.getFullStackTrace(ex));
                                } finally {
                                    if (sender != null) {
                                        pool.Return(sender);
                                    }
                                }
                            } else {
                                logger.warn("get string is not fit"+obj_gaterTime);
                            }
                        }

                        logger.info("finish fileName = " + file.getAbsolutePath());
                        File finishFile = new File(file.getAbsolutePath().replaceAll("\\\\", "/") + ".ok");
                        file.renameTo(finishFile);
                        logger.info("finished redo fileName = " + finishFile.getAbsolutePath());
                    } catch (FileNotFoundException ex) {
                        logger.error(ex.getMessage(), ex);
                    } catch (IOException ex) {
                        logger.error(ex.getMessage(), ex);
                    } finally {
                        try {
                            reader.close();
                        } catch (IOException e) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }finally{
            logger.info(dataType+" send to mq "+GlobalParas.CONSUMER_DATATYPE_TOPIC_MAP.get(dataType)+" finished");
        }
    }

    public static ArrayList<File> getFile(File dir) {
        ArrayList<File> fileList = new ArrayList<>();
        try {
            if (!dir.isDirectory() || !dir.exists()) {
                dir.mkdirs();
                logger.warn("path = " + dir + "is not exist");
            }
            File[] yfiles = dir.listFiles();
            if (yfiles.length > 0) {
                for (File yfile : yfiles) {
                    String fileName = yfile.getName();
                    if (!fileName.endsWith(".ok")) {//表明未处理完
                        fileList.add(yfile);
                        logger.info("get fileName = " + fileName);
                    }
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return fileList;
    }
}
